package cn.edu.hit.core.combine;

import cn.edu.hit.core.transfer.ReduceTransfer;
import cn.edu.hit.core.transfer.TransferLoader;
import cn.edu.hit.core.Document;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public final class ReduceAdapter extends CombineAdapterHelper {
    @Override
    public boolean accept(String type) {
        return "reduce".equalsIgnoreCase(type);
    }

    @Override
    protected Object combineAndCalculate(List<Document> channelDocs, Document configure, String method, Map<String, Object> arg) {
        List<Object> channelValues = channelDocs.stream()
                .map(doc -> doc.get(VALUE))
                .collect(Collectors.toList());
        Stream<Object> tStream = Stream.empty();

        ReduceTransfer transfer = TransferLoader.getReduce(method);
        if (transfer.canParallel()) {
            tStream = channelValues.parallelStream();
        } else {
            tStream = channelValues.stream();
        }
//            Object finalTransferArgObj = transferArg;
        return tStream.reduce((t1, t2) -> transfer.merge(t1, t2, configure)).orElse(null);
    }
}
